package com.sanerzone.smscenter.biz.listener;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.sanerzone.common.support.utils.FstObjectSerializeUtil;
import com.sanerzone.smscenter.biz.entity.SmsReceive;
import com.sanerzone.smscenter.biz.mapper.SmsReceiveMapper;
import com.sanerzone.smscenter.biz.message.SMSMOMessage;
import com.sanerzone.smscenter.biz.message.SMSReplyMessage;
import com.sanerzone.smscenter.common.tools.MQHelper;
import com.sanerzone.smscenter.common.tools.SpringContextHelper;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;

@Service
public class SMSReplySaveListener implements MessageListenerConcurrently {


    public static Logger logger = LoggerFactory.getLogger(SMSReplySaveListener.class);

    @Autowired
    public MQHelper mQUtils;

    @Autowired
    private SmsReceiveMapper smsReceiveMapper;

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        int index =0;
        try{
            for (MessageExt msg: msgs) {//遍历
                if (msg.getReconsumeTimes() > 2 ) {
                    continue;
                }
                logger.info("reply listener recv message: topic:{}, tags:{}, msgid:{}, key:{}", msg.getTopic(), msg.getTags(),
                        msg.getMsgId(), msg.getKeys());
                SMSMOMessage smsmoMessage= (SMSMOMessage)FstObjectSerializeUtil.read(msg.getBody());
                smsmoMessage.setMsgid(msg.getMsgId());
                if(smsmoMessage != null ){
                    smsReceiveMapper.insertMOAccountSmsReceive( smsmoMessage );
                } else {
                    logger.info("reply listener recv message: topic:{}, tags:{}, msgid:{}, key:{}", msg.getTopic(), msg.getTags(),
                            msg.getMsgId(), msg.getKeys());
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
        }catch (Exception e){
            logger.error("INSMSReply-ERR", e);
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
